package com.bd03.streaminglearn.day0325
import java.sql.DriverManager

import com.typesafe.config.ConfigFactory
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WordCount2 {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)//屏蔽日志
    // 创建一个具有两个工作线程（working thread）并且批次间隔为 2 秒的本地 StreamingContext。
    // master 需要 2 个核，以防止饥饿情况（starvation scenario）。
    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      .setMaster("local[*]")
    val ssc: StreamingContext = new  StreamingContext(conf,Seconds(2))
    val config = ConfigFactory.load()

    val dStream: DStream[String] = ssc.socketTextStream("hdp01",9999)

    val res: DStream[(String, Int)] = dStream.flatMap(t => {
      t.split(" ")
    }).map((_, 1)).reduceByKey(_ + _).repartition(2)
    res.print()


    //带历史状态的wordcount
    res.foreachRDD(rdd=>{
      println("-------------------------")
      rdd.foreachPartition(t=>{
        val url = "jdbc:mysql://hdp03:3306/spark?useUnicode=true&characterEncoding=utf-8"
        val con = DriverManager.getConnection(url,"root","root")
        val sql = "insert into wordcount values(?,?)"
        val sqlSerach = "select num from wordcount where word = ?"
        val sqlUpdate = "update wordcount  set num = ? where word = ?"
        val ps = con.prepareStatement(sql)
        val ps2 = con.prepareStatement(sqlSerach)
        val ps3 = con.prepareStatement(sqlUpdate)
        t.foreach(m=>{
          ps2.setString(1,m._1)
          val resSerach = ps2.executeQuery()
          //a 1 b 4
          if(resSerach.next()){//有值就改值,
            val i = resSerach.getInt("num")
            val newCount = i+m._2
            ps3.setInt(1,newCount)
            ps3.setString(2,m._1)
            ps3.executeUpdate()
          }else{// 没有值则添加值
            ps.setString(1,m._1)
            ps.setInt(2,m._2)
            ps.executeUpdate()
          }

        })
        ps.close()
        ps2.close()
        ps3.close()
        con.close()
      })

    })


    ssc.start()//启动
    ssc.awaitTermination() //等待被终止
  }

}
